/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package DAO;

import Common.Constants;
import Common.GeneralHandling;
import POJO.CounterPOJO;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.*;
import org.apache.thrift.TException;

/**
 *
 * @author node1
 */
public class CounterDAO {
    public static ByteBuffer colParent = ByteBuffer.wrap("Counter".getBytes());
    public static ByteBuffer keyCounter = ByteBuffer.wrap("Counter".getBytes());
    public static ByteBuffer colNameTag = ByteBuffer.wrap("counterTag".getBytes());
    public static ByteBuffer colNameStatus = ByteBuffer.wrap("counterStatus".getBytes());
    
    public static String colParentS = "Counter";
    public static String keyCounterS = "Counter";
    public static String colNameTagS = "counterTag";
    public static String colNameStatusS = "counterStatus";
    
    private static Cassandra.Client client;
    private static CounterDAO instance;

    public static CounterDAO getInstance(Cassandra.Client _client){
        if (instance != null)
            return instance;
        instance = new CounterDAO();
        client = _client;
        return instance;
    }
 
    public static CounterPOJO getCounter(Cassandra.Client client){
        CounterPOJO counter = new CounterPOJO();
        try {
            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameStatus);
            predicate.addToColumn_names(colNameTag);
            ColumnParent parent = new ColumnParent(colParentS);
            List<ColumnOrSuperColumn> result = client.get_slice(keyCounter, parent, predicate, Constants.CL);
            int counterStatus = GeneralHandling.toInt(result.get(0).column.value);
            int counterTag = GeneralHandling.toInt(result.get(1).column.value);
            counter.counterStatus = counterStatus;
            counter.counterTag = counterTag;
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return counter;
    }
    

    public static int getCounterStatus(Cassandra.Client client){
        int result = 0;
        try {
            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameTag);
            ColumnParent parent = new ColumnParent(colParentS);
            List<ColumnOrSuperColumn> colOrSupCol = client.get_slice(keyCounter, parent, predicate, Constants.CL);
            result = GeneralHandling.toInt(colOrSupCol.get(0).column.value);

        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }    
    
    public static int getCounterTag(Cassandra.Client client){
        int result = 0;
        try {
            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameStatus);
            ColumnParent parent = new ColumnParent(colParentS);
            List<ColumnOrSuperColumn> colOrSupCol = client.get_slice(keyCounter, parent, predicate, Constants.CL);
            result = GeneralHandling.toInt(colOrSupCol.get(0).column.value);

        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }     
    
     public static boolean Inscrease(ByteBuffer colName, Cassandra.Client client){
        boolean result = false;
        try {            
            ColumnPath path = new ColumnPath(CounterDAO.colParentS);
            ColumnParent parent = new ColumnParent(CounterDAO.colParentS);            
            path.column = colName;
            
            ColumnOrSuperColumn counter = client.get(keyCounter, path, Constants.CL);
            Column counterCol = counter.column;
            long counterCurrent = GeneralHandling.toLong(counterCol.value);            
            counterCurrent += 1;            
            client.insert(keyCounter, parent, GeneralHandling.createCol(colName, GeneralHandling.toStringFromLong(counterCurrent)), Constants.CL);
            result = true;
        } catch (NotFoundException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(UserDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }    
    
    public static boolean Decrease(ByteBuffer colName, Cassandra.Client client)
    {
        boolean result = false;
        try {
            ColumnPath path = new ColumnPath(CounterDAO.colParentS);
            ColumnParent parent = new ColumnParent(CounterDAO.colParentS);            
            path.column = colName;
            
            ColumnOrSuperColumn counter = client.get(keyCounter, path, Constants.CL);
            Column counterCol = counter.column;
            long counterCurrent = GeneralHandling.toLong(counterCol.value);
            counterCurrent -= 1;
            client.insert(keyCounter, parent, GeneralHandling.createCol(colName, GeneralHandling.toStringFromLong(counterCurrent)), Constants.CL);
            result = true;
        } catch (NotFoundException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(UserDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }
    
    public void setCounterGlobal()
    {
        try {
            SlicePredicate predicateStatus = new SlicePredicate();
            predicateStatus.addToColumn_names(StatusDAO.colNameCreateDate);
            
            ColumnParent parentStatus = new ColumnParent(StatusDAO.colParentS);
            KeyRange range = new KeyRange();
            range.start_key = Common.GeneralHandling.toByteBuffer("");
            range.end_key = Common.GeneralHandling.toByteBuffer("");
            range.count = Short.MAX_VALUE;
            List<KeySlice> results = client.get_range_slices(parentStatus, predicateStatus, range, Common.Constants.CL);       
            int counterStatus = results.size();
            
            SlicePredicate predicateTag = new SlicePredicate();
            predicateTag.addToColumn_names(TagDAO.colNameCreateDate);
            ColumnParent parentTag = new ColumnParent(TagDAO.colParentListStatusS);
            results = client.get_range_slices(parentTag, predicateTag, range, Common.Constants.CL);       
            int counterTag = results.size();            
            
            ColumnParent parentCounter = new ColumnParent(CounterDAO.colParentS);
            client.insert(keyCounter, parentCounter, GeneralHandling.createCol(colNameStatus, Integer.toString(counterStatus)), Constants.CL);
            client.insert(keyCounter, parentCounter, GeneralHandling.createCol(colNameTag, Integer.toString(counterTag)), Constants.CL);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        catch (UnsupportedEncodingException ex) {
            Logger.getLogger(CounterDAO.class.getName()).log(Level.SEVERE, null, ex);
        }       
    }
}


